package org.devgframwork.rabbitmq.connection;

import org.devgframwork.rabbitmq.DevgRabbitProperties;
import org.devgframwork.rabbitmq.DevgShutdownListener;
import org.devgframwork.rabbitmq.RabbitComponent;
import com.rabbitmq.client.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.LinkedList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeoutException;

public class DefaultConnectionFactory extends AbstractConnectionFactory {

    private final Logger logger = LoggerFactory.getLogger(DefaultConnectionFactory.class);

    private static final boolean defaultRabbitAutomaticRecoveryEnabled = true;

    private static final int defaultRabbitNetworkRecoveryInterval = 10000;

    private static final int defaultRabbitRequestedHeartbeat = 60;

    private static final int defaultRabbitConnectionTimeout = 30000;

    private static final int defaultRabbitHandshakeTimeout = 30000;

    private static final int defaultRabbitMaxChannelSize = 20;

    public static int rabbitThreadPoolSize = 0;

    private ExecutorService executorService;

    private AddressResolver addressResolver;

    private DevgRabbitProperties devgRabbitProperties;

    private final LinkedList<Channel> channelList = new LinkedList<>();

    private final LinkedList<Channel> transactionalChannelList = new LinkedList<>();

    private Connection connection;

    public DefaultConnectionFactory(DevgRabbitProperties devgRabbitProperties) {
        super(newRabbitConnectionFactory(devgRabbitProperties));
        this.devgRabbitProperties = devgRabbitProperties;
    }


    @Override
    public int getChannelSize() {
        return channelList.size();
    }

    public void setAddressResolver(AddressResolver addressResolver) {
        this.addressResolver = addressResolver;
    }

    private static ConnectionFactory newRabbitConnectionFactory(DevgRabbitProperties devgRabbitProperties) {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(devgRabbitProperties.getRabbitHost());
        connectionFactory.setPort(devgRabbitProperties.getRabbitPort());
        connectionFactory.setUsername(devgRabbitProperties.getRabbitUsername());
        connectionFactory.setPassword(devgRabbitProperties.getRabbitPassword());
        String rabbitVirtualHost = devgRabbitProperties.getRabbitVirtualHost();
        if (null != rabbitVirtualHost && !rabbitVirtualHost.equals("")) {
            connectionFactory.setVirtualHost(rabbitVirtualHost);
        }
        Integer networkRecoveryInterval = devgRabbitProperties.getRabbitNetworkRecoveryInterval();
        Boolean automaticRecoveryEnabled = devgRabbitProperties.getRabbitAutomaticRecoveryEnabled();
        Integer requestedHeartbeat = devgRabbitProperties.getRabbitRequestedHeartbeat();
        Integer connectionTimeout = devgRabbitProperties.getRabbitConnectionTimeout();
        Integer handshakeTimeout = devgRabbitProperties.getRabbitHandshakeTimeout();
        Integer poolSize = devgRabbitProperties.getRabbitThreadPoolSize();
        if (null != poolSize) {
            rabbitThreadPoolSize = poolSize.intValue();
        }
        connectionFactory.setAutomaticRecoveryEnabled(automaticRecoveryEnabled == null ? defaultRabbitAutomaticRecoveryEnabled : automaticRecoveryEnabled);
        connectionFactory.setNetworkRecoveryInterval(networkRecoveryInterval == null ? defaultRabbitNetworkRecoveryInterval : networkRecoveryInterval);
        connectionFactory.setRequestedHeartbeat(requestedHeartbeat == null ? defaultRabbitRequestedHeartbeat : requestedHeartbeat);
        connectionFactory.setConnectionTimeout(connectionTimeout == null ? defaultRabbitConnectionTimeout : connectionTimeout);
        connectionFactory.setHandshakeTimeout(handshakeTimeout == null ? defaultRabbitHandshakeTimeout : handshakeTimeout);
        return connectionFactory;
    }


    @Override
    public Connection createConnection() {
        try {
            if (null != connection && connection.isOpen()) {
                return connection;
            }
            DevgShutdownListener shutdownListener = null;
            try {
                shutdownListener = RabbitComponent.getApplicationContext().getBean(DevgShutdownListener.class);
            } catch (Exception ex) {
            }
            synchronized (this) {
                if (null != connection && connection.isOpen()) {
                    return connection;
                }
                if (this.addressResolver != null) {
                    connection = this.getRabbitConnectionFactory().newConnection(addressResolver);
                    if (null != shutdownListener) {
                        connection.addShutdownListener(shutdownListener);
                    }
                    return connection;
                }
                connection = this.getRabbitConnectionFactory().newConnection();
                if (null != shutdownListener) {
                    connection.addShutdownListener(shutdownListener);
                }
                return connection;
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        } catch (TimeoutException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public Channel createChannel(ChannelType channelType) {
        Channel channel = null;
        LinkedList<Channel> targetList = channelList;
        String targetChannelType = channelType.getType();
        if (targetChannelType.equals(ChannelType.TRANSACTIONAL.getType())) {
            targetList = transactionalChannelList;
        }
        synchronized (targetList) {
            while (!targetList.isEmpty()) {
                channel = targetList.removeFirst();
                if (channel.isOpen()) {
                    break;
                }
            }
            try {
                if (null == channel) {
                    channel = createConnection().createChannel();
                }
                return channel;
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }

    @Override
    public void closeChannelAndConnection(Channel channel, ChannelType channelType) {
        if (channel != null) {
            try {
                Integer maxChannelSize = devgRabbitProperties.getRabbitMaxChannelSize();
                if (null == maxChannelSize || maxChannelSize <= 0) {
                    maxChannelSize = defaultRabbitMaxChannelSize;
                }
                LinkedList<Channel> targetList = channelList;
                String targetChannelType = channelType.getType();
                if (targetChannelType.equals(ChannelType.TRANSACTIONAL.getType())) {
                    targetList = transactionalChannelList;
                }
                synchronized (targetList) {
                    if (targetList.size() >= maxChannelSize) {
                        channel.close();
                        if (targetList.contains(channel)) {
                            targetList.remove(channel);
                        }
                        return;
                    }
                    if (!targetList.contains(channel)) {
                        targetList.addLast(channel);
                    }
                }
            } catch (AlreadyClosedException var2) {
            } catch (ShutdownSignalException var4) {
                if (!isNormalShutdown(var4)) {
                    logger.debug("Unexpected exception on closing RabbitMQ Channel", var4);
                }
            } catch (Exception var5) {
                logger.debug("Unexpected exception on closing RabbitMQ Channel", var5);
            }
        }
    }

    public static boolean isNormalShutdown(ShutdownSignalException sig) {
        Method shutdownReason = sig.getReason();
        return shutdownReason instanceof AMQP.Connection.Close && 200 == ((AMQP.Connection.Close) shutdownReason).getReplyCode() && "OK".equals(((AMQP.Connection.Close) shutdownReason).getReplyText());
    }

}
